package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;
import sun.security.pkcs11.P11Util;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public abstract class DimAsyncRichFunction<T> extends RichAsyncFunction<T, T> implements DimAsyncFunction<T> {
    private AsyncConnection hbaseAsyncConnection;
    private StatefulRedisConnection<String, String> redisAsyncConnection;

    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseAsyncConnection = HBaseUtil.getAsyncConnection();
        redisAsyncConnection = RedisUtil.getAsyncConnection();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeAsyncConnection(hbaseAsyncConnection);
        RedisUtil.closeAsyncConnection(redisAsyncConnection);
    }

    @Override
    public void asyncInvoke(T bean, ResultFuture<T> resultFuture) throws Exception {
        String rowKey = getRowKey(bean);
        String tableName = getTableName();
        // 第一步 从redis中读取维度数据
        CompletableFuture.<JSONObject>supplyAsync(
                        new Supplier<JSONObject>() {
                            @Override
                            public JSONObject get() {
                                JSONObject dimJsonObj = RedisUtil.readDimAsync(redisAsyncConnection, tableName, rowKey);
                                return dimJsonObj;
                            }
                        }
                )
                // 第二步 如果读取不到,再从hbase中读取维度数据
                .<JSONObject>thenApplyAsync(
                        new Function<JSONObject, JSONObject>() {
                            @Override
                            public JSONObject apply(JSONObject dimFromRedis) {
                                JSONObject dimFromHBase = dimFromRedis;
                                if (dimFromHBase == null) {
                                    // 读不到去hbase读取维度数据
                                    dimFromHBase = HBaseUtil.getAsyncRow(hbaseAsyncConnection, Constant.HBASE_NAMESPACE, tableName, rowKey, JSONObject.class, false);
                                    if (dimFromHBase.size() > 0) {
                                        // 写入到redis
                                        RedisUtil.writeDimAsync(redisAsyncConnection, tableName, rowKey, dimFromHBase);
                                    }
                                }
                                return dimFromHBase;
                            }
                        }
                )
                // 第三步 读取到维度数据后 进行维度关联
                .thenAcceptAsync(
                        new Consumer<JSONObject>() {
                            @Override
                            public void accept(JSONObject dimFromHbase) {
                                if (dimFromHbase.size() > 0) {
                                    addDims(bean, dimFromHbase);
                                }
                                // 将数据写出
                                resultFuture.complete(Collections.singleton(bean));
                            }
                        }
                );
    }
}
